package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
//import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * 电子工业出版社的2017年4月第一版《Spark GraphX in Action》第四章
  * Created by Administrator on 2017/4/21 0021.
  * 构建图4.3中的示例图。
  */
object codelist4_1 {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val myVertices = sc.makeRDD(Array((1L, "Ann"),(2L, "Bill"), (3L, "Charles"), (4L, "Diane"), (5L, "Went to gym this morning")))
    val myEdges = sc.makeRDD(Array(Edge(1L, 2L, "is-friends-with"), Edge(2L, 3L, "is-friends-with"),
      Edge(3L, 4L, "is-friends-with"), Edge(4L, 5L, "Likes-status"), Edge(3L, 5L, "Wrote-status")))
    val myGraph = Graph(myVertices, myEdges)
    val list_vertices = myGraph.vertices.collect()
    println("展示图的顶点:")
    for(vertice <- list_vertices){
      println(vertice)
    }
    val list_edges = myGraph.edges.collect()
    println("展示图的边：")
    for(edge <- list_edges){
      println(edge)
    }
    val list_triplets = myGraph.triplets.collect()
    println("展示图的边三元组:")
    for(triplet <- list_triplets){
      println(triplet)
    }


    //简单的图转换
    //对每一条满足这两个条件的边增加属性值：1,属性中包含"is-friends-with"的边；2，关系的源顶点中包含字母a。
    //val list_cv1 = myGraph.mapTriplets(t =>(t.attr, t.attr=="is-friends-with" && t.srcAttr.toLowerCase.contains("a"))).triplets.collect
    //如果没有声明匿名函数的返回类型，Scala会根据函数中的返回的结果推断出返回类型。如果我们希望Scala编译器双重检查类型(以及代码标注的类型)
    //应该将val list_cv1 = myGraph.mapTriplets...这一行代码改写为如下
    val list_cv1 = myGraph.mapTriplets((t =>(t.attr, t.attr=="is-friends-with" && t.srcAttr.toLowerCase.contains("a"))) : (EdgeTriplet[String,String] => Tuple2[String,Boolean]) ).triplets.collect
    println("展示修改后的图的边三元组:")
    for(triplet <- list_cv1){
      println(triplet)
    }
    //与允许转变Edge类的mapTriplets()类似，函数mapVertices允许我们直接转变Vertex类。我们可以自己查看官网或者查看之前自己写的其它代码来查看mapVertices类。



    //下面一段代码会统计每个顶点的“出度”——即对于每个顶点而言，离开该顶点的边的条数。为了完成该操作，我们会间接处理每条边以及与边
    //相关联的源顶点和目标顶点。相比于直接统计从每个顶点出发的边条数，我们会让每条边发出消息到关联的源顶点。这两种方法是等效的。汇总这些
    //消息后，就可以得到我们想要的答案。
    val list_cv2 = myGraph.aggregateMessages[Int](_.sendToSrc(1), _ + _).collect
    println("每个顶点的出度:")
    for(t1 <- list_cv2){
      println(t1)
    }
    //需要注意的是，由于顶点#5不包含任何出边，它接收不到任何消息，所以它不会出现在结果中。
    //collect之前，aggregateMessages方法的返回值是VertexRDD[INT]对象。
    //关于aggregateMessages的详细解释，可以参看电子工业出版社的2017年4月第一版《Spark GraphX in Action》的章节4.2.2/Page73



    //前面的aggregateMessages的操作结果，如下
    /*
    * 每个顶点的出度:(4,1),(1,1),(3,2),(2,1)
    * */
    //这原始的VertexId并不容易理解，我们需要为顶点添上可读的名字。
    //使用join()方法
    val list_cv3 = myGraph.aggregateMessages[Int](_.sendToSrc(1), _ + _).join(myGraph.vertices).collect
    println("每个顶点的出度(每个VertexId后面跟两个值，第一个是出度，第二个是VertexId的属性——名字):")
    for(t2 <- list_cv3){
      println(t2)
    }
    /*
    显示的结果：
    (4,(1,Diane))
    (1,(1,Ann))
    (3,(2,Charles))
    (2,(1,Bill))*/



    //上述结果看起来不错，可是如果我们不需要VertexId了，并且在输出的时候，想让名字在前面，出度在后面，该如何操作呢？
    val list_cv4 = myGraph.aggregateMessages[Int](_.sendToSrc(1), _ + _).join(myGraph.vertices).map(_._2.swap).collect
    println("可视化的输出结果:")
    for(t2 <- list_cv4){
      println(t2)
    }


    //很好，现在我们想输出所有的名字，就算没有出度。
    val list_cv5 = myGraph.aggregateMessages[Int](_.sendToSrc(1), _ + _).rightOuterJoin(myGraph.vertices).map(_._2.swap).collect
    println("可视化的输出结果(包括没有出度的#5):")
    for(t2 <- list_cv5){
      println(t2)
    }
    /*(Diane,Some(1))
    (Ann,Some(1))
    (Charles,Some(2))
    (Went to gym this morning,None)
    (Bill,Some(1))*/
    //我们注意到，数字外面包了Some，该有0的地方出现了None。



    //为了除去Some和用0代替None，我们应当编写如下的代码。
    val list_cv6 = myGraph.aggregateMessages[Int](_.sendToSrc(1), _ + _).rightOuterJoin(myGraph.vertices).map(x => (x._2._2, x._2._1.getOrElse(0))).collect
    println("优化后的可视化的输出结果(包括没有出度的#5):")
    for(t2 <- list_cv6){
      println(t2)
    }
    /*
    结果如下：
    (Diane,1)
    (Ann,1)
    (Charles,2)
    (Went to gym this morning,0)
    (Bill,1)
     */

    sc.stop()
  }
}
